package com.cnd.scm.rabbitmq.common;

import com.cnd.scm.common.utils.JsonUtils;
import com.cnd.scm.redis.cache.util.RedisUtil;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.stereotype.Service;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.UUID;

/**
 * @version V1.0
 * @ClassName: DbMessageSender
 * @Description:
 * @author: 李启联
 * @date: 2021-03-22 17:19
 * @Copyright: Copyright (c), 2021
 */

@Service
public class DbMessageSender implements MessageSender{


    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisUtil redisUtil;


    @Autowired
    private JdbcTemplate jdbcTemplate;


    @Override
    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    /**
     * 向发送消息容器增加要发送的消息
     *
     * @param messageLog
     */
    @Override
    public void storeMessage(MessageLog messageLog) {
        String sql = "INSERT msg_log(`message_id`,`message`,`retry_num`,`exchange`,`routing_key`,`status`,`create_time`,`update_time`) VALUES(?,?,?,?,?,?,?,?)";
        int update = jdbcTemplate.update(sql,
                new PreparedStatementSetter() {
                    @Override
                    public void setValues(PreparedStatement preparedStatement) throws SQLException {
                        preparedStatement.setString(1,messageLog.getMessageId());
                        preparedStatement.setString(2, JsonUtils.toJsonString(messageLog.getMessage()));
                        preparedStatement.setInt(3,0);
                        preparedStatement.setString(4,messageLog.getExchange());
                        preparedStatement.setString(5, messageLog.getRoutingKey());
                        preparedStatement.setInt(6,MessageStatusEnum.SEND.getCode());
                        preparedStatement.setTimestamp(7,new java.sql.Timestamp(System.currentTimeMillis()));
                        preparedStatement.setTimestamp(8,new java.sql.Timestamp(System.currentTimeMillis()));
                    }
                }
        );
     }

    /**
     * 更新保存在容器中投递消息的状态
     *
     * @param status 消息状态
     * @param msgId  消息ID
     */
    @Override
    public void updateStatus(MessageStatusEnum status, String msgId) {
        String sql = "UPDATE msg_log set `status` = ? ,`update_time`=? where message_id=?";
        int update = jdbcTemplate.update(sql, new PreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement preparedStatement) throws SQLException {
                preparedStatement.setInt(1,status.getCode());
                preparedStatement.setTimestamp(2, new java.sql.Timestamp(System.currentTimeMillis()));
            }
        });
    }

    /**
     * 删除发送容器中的一条消息
     *
     * @param id
     */
    @Override
    public void delMessage(String id) {

    }

    /**
     * 发送消息
     *
     * @param message
     * @return
     */
    @Override
    public DetailResponse send(Object message) {
        String id = UUID.randomUUID().toString().replace("-","").toUpperCase();
        return send(new MessageLog(id, message,0,rabbitTemplate.getExchange(),rabbitTemplate.getRoutingKey()));
    }

    /**
     * 发送消息
     *
     * @param messageLog
     * @return
     */
    @Override
    public DetailResponse send(MessageLog messageLog) {
        try {
            storeMessage(messageLog);
            rabbitTemplate.correlationConvertAndSend(messageLog.getMessage(), new CorrelationData(String.valueOf(messageLog.getMessageId())));
        } catch (Exception e) {
            return new DetailResponse(false, "", "");
        }
        return new DetailResponse(true, "", "");
    }

    /**
     * 开启重试发送机制
     */
    @Override
    public void retry() {
        new Thread(() ->{
            while (true) {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                long now = System.currentTimeMillis();
                System.out.println("DbMessageSender startRetry..." );

            }
        }).start();

    }
}